package cn.ucox.web.ms.packer;

import cn.ucox.web.ms.Constant;
import cn.ucox.web.ms.components.cache.TableMetaInfoCache;
import cn.ucox.web.ms.polling.IFileLoader;
import cn.ucox.web.ms.polling.SyncAnchor;
import cn.ucox.web.ms.polling.SyncData;
import cn.ucox.web.ms.redis.RedisClient;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Base64Utils;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;

import java.io.UnsupportedEncodingException;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import static cn.ucox.web.ms.Constant.*;

/**
 * 业务数据处理器
 *
 * @author chenw
 * @create 2017-05-11 16:01
 * @email javacspring@gmail.com
 */
public class TransactionDataProcessor implements Runnable {

    public static final Logger logger = LoggerFactory.getLogger(TransactionDataProcessor.class);


    private List<SyncData> mSyncData;
    private IFileLoader mFileLoader;//文件加载器
    private SyncAnchor mSyncAnchor;
    private static TableMetaInfoCache tableMetaInfoCache;

    private static RedisClient redisClient;

    static {
        redisClient = RedisClient.instance();
        tableMetaInfoCache = TableMetaInfoCache.getInstance();
    }

    public TransactionDataProcessor(SyncAnchor anchor,
                                    IFileLoader fileLoader,
                                    List<SyncData> syncData) {
        this.mSyncAnchor = anchor;
        this.mSyncData = syncData;
        this.mFileLoader = fileLoader;
    }

    @Override
    public void run() {
        long dataSize = 0;
        long totalSize = 0;
        Jedis jedis = null;
        try {
            jedis = redisClient.getJedis();
            Pipeline pipeline = jedis.pipelined();
            for (SyncData syncData : mSyncData) {
                JSONObject rowData = new JSONObject(4) {{
                    put("tab", syncData.getTable());
                    put("act", syncData.getAction());
                    put("ctm", syncData.getTime());
                    put("id", syncData.getSyncId());
                }};
                //是否为附件数据
                if (syncData.isAttachmentData()) {
                    String tabName = syncData.getTable();
                    //原始同步表行记录
                    JSONObject rowDataJson = JSON.parseObject(syncData.getData());
                    String attCol = tableMetaInfoCache.getTableAttachmentColumn(tabName);
                    String fileUrl = rowDataJson.getString(attCol);

                    //执行删除动作时，附件数据不应该再次下载，只需要将附件元信息同步进去即可
                    if (ACTION_DELETE == syncData.getAction()) {
                        rowDataJson.put("__att_col_name__", attCol);
                        rowDataJson.put("__att_content__", "");
                    } else {
                        // 调用FS客户端进行文件获取   外网
                        JSONObject attData = mFileLoader.getFile(fileUrl);
                        //调用接口进行FS文件获取     信息网
                        //JSONObject attData = mFileLoader.getFileByInterface(fileUrl);
                        if (null != attData) {
                            rowDataJson.put("__att_col_name__", attCol);
                            rowDataJson.put("__att_content__", attData);
                        }
                    }
                    rowData.put("data", Base64Utils.encodeToString(rowDataJson.toString().getBytes("UTF-8")));
                    rowData.put("att", true);
                    byte[] content = (rowData.toJSONString() + "\n").getBytes(Charset.forName("UTF-8"));
                    jedis.hset(SYNC_DATA_BYTE, syncData.getTime().getBytes(), content);
                    dataSize += content.length;
                } else {//普通文本类数据
                    rowData.put("att", false);
                    rowData.put("data", Base64Utils.encodeToString(syncData.getData().getBytes("UTF-8")));
                    byte[] content = (rowData.toJSONString() + "\n").getBytes(Charset.forName("UTF-8"));
                    jedis.hset(SYNC_DATA_BYTE, syncData .getTime().getBytes(), content);
                    dataSize += content.length;
                }
            }

            jedis.hincrBy(SYNC_ANCHOR, SYNC_ANCHOR_TRANSACTION_DATA_SIZE, dataSize);
            pipeline.sync();
        } catch (Exception ex) {
            logger.error("写入子任务数据错误:{}", ex.getMessage(), ex);
        } finally {
            if (null != jedis) {
                try {
                    jedis.close();
                } catch (Exception ex) {
                    logger.error("关闭jedis客户端错误", ex);
                }
            }
        }

        try {
            jedis = redisClient.getJedis();
            Pipeline pipeline = jedis.pipelined();
            long doneTask = jedis.hincrBy(SYNC_ANCHOR, SYNC_ANCHOR_TASK_DONE_NUM, 1L);
            logger.debug("统计信息:数据大小:{},完成任务:{},总任务数:{}", dataSize, doneTask, mSyncAnchor.getTotalTaskNum());
            //最后一个子任务
            if (doneTask == mSyncAnchor.getTotalTaskNum()) {

                Map<byte[], byte[]> redisData = jedis.hgetAll(SYNC_DATA_BYTE);
                List<byte[]> keys = redisData.keySet().stream().sorted((byte[] o1, byte[] o2) -> {
                    try {
                        return new java.lang.String(o1, "UTF-8").compareTo(new java.lang.String(o2, "UTF-8"));
                    } catch (UnsupportedEncodingException e) {
                        logger.error("decoding data key to string error", e);
                        return 1;
                    }
                }).collect(Collectors.toList());

                totalSize = Long.valueOf(jedis.hget(SYNC_ANCHOR, SYNC_ANCHOR_TRANSACTION_DATA_SIZE));
                //写入事务头部
                jedis.lpush(SYNC_QUEUE_BYTE, getTransactionDataHeadByte(mSyncAnchor.getTransactionId(),
                        totalSize,
                        mSyncAnchor.getTransactionDataRows()));
                //逐条写入数据并清空临时数据集合
                for (byte[] key : keys) {
                    byte[] data = redisData.get(key);
                    jedis.lpush(SYNC_QUEUE_BYTE, data);
                    jedis.hdel(SYNC_DATA_BYTE, keys.toArray(new byte[][]{}));
                }
                //写入事务尾部
                jedis.lpush(SYNC_QUEUE_BYTE,
                        getTransactionDataTailByte(mSyncAnchor.getTransactionId(),
                                totalSize,
                                mSyncAnchor.getTransactionDataRows()));
                /*jedis.hmset(SYNC_ANCHOR, new HashMap<String, String>() {{
                    //put(SYNC_ANCHOR_TOTAL_TASK_NUM, "0");
                    //put(SYNC_ANCHOR_TASK_DONE_NUM, "0");
                    put(SYNC_ANCHOR_TRANSACTION_DONE, "Y");
                    //put(SYNC_ANCHOR_TRANSACTION_DATA_ROWS, "0");
                    //put(SYNC_ANCHOR_TRANSACTION_DATA_SIZE, "0");
                }});*/
                logger.debug("-子线程[{}]处理结束,总任务:{},已完成:{}", Thread.currentThread().getName(), mSyncAnchor.getTotalTaskNum(), doneTask);
                logger.debug("事务结束");
            } else {

                logger.debug("+子线程[{}]处理结束,总任务:{},已完成:{}", Thread.currentThread().getName(), mSyncAnchor.getTotalTaskNum(), doneTask);
            }
            pipeline.sync();
        } catch (Exception ex) {
            logger.error("汇总事务数据错误:{}", ex.getMessage(), ex);
        } finally {
            try {
                jedis.close();
            } catch (Exception ex) {
                logger.error("关闭jedis客户端错误", ex);
            }
        }
    }

    private byte[] getTransactionDataHeadByte(long transactionId, long size, int items) {
        //##DEX_HEAD`tid`size`items
        return String.format("%s`%d`%d`%s", Constant.SYNC_QUEUE_HEAD_FLAG, transactionId, items, size).getBytes();
    }

    private byte[] getTransactionDataTailByte(long transactionId, long size, int items) {
        //##DEX_TAIL`tid`size`items
        return String.format("%s`%d`%d`%s", Constant.SYNC_QUEUE_TAIL_FLAG, transactionId, items, size).getBytes();
    }
}
